package org.homonoia.controller.queue;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.ObjectOutputStream;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.AbstractMap;
import java.util.AbstractMap.SimpleEntry;
import java.util.Date;
import java.util.Properties;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.log4j.Logger;
import org.homonoia.model.query.Query;
import org.homonoia.model.query.QueryID;
import org.homonoia.model.query.QueryResults;
import org.homonoia.model.query.QueryStatus;
import org.homonoia.model.task.Task;
import org.homonoia.model.task.TaskResults;

public class JobQueue {
    private static final Logger log = Logger.getLogger(JobQueue.class);

    volatile private Queue<QueuedJob> waitingJobs;
    volatile private Queue<QueuedJob> runningJobs;
    volatile private Queue<QueuedJob> completedJobs;

    private Properties configProps;

    public JobQueue(final Properties configProps) {
	this.configProps = configProps;

	this.waitingJobs = new ConcurrentLinkedQueue<QueuedJob>();
	this.runningJobs = new ConcurrentLinkedQueue<QueuedJob>();
	this.completedJobs = new ConcurrentLinkedQueue<QueuedJob>();
    }

    public final QueryID newJob(final Query job) {
	QueuedJob newJob = new QueuedJob(job);
	newJob.generateTasks(configProps);

	synchronized (waitingJobs) {
	    waitingJobs.add(newJob);
	}

	return newJob.getJobID();
    }

    public final QueryStatus getJobStatus(final QueryID id) {
	QueryStatus status = null;
	QueuedJob job = null;

	if (status == null) {
	    job = findJobInQueue(waitingJobs, id);
	    if (job != null) {
		status = job.getJobStatus();
	    }
	}

	if (status == null) {
	    job = findJobInQueue(runningJobs, id);
	    if (job != null) {
		status = job.getJobStatus();
	    }
	}

	if (status == null) {
	    job = findJobInQueue(completedJobs, id);
	    if (job != null) {
		status = job.getJobStatus();
	    }
	}

	return status;
    }

    public final QueryResults getJobResults(final QueryID id) {
	QueryResults history = null;
	QueuedJob job = null;

	if (history == null) {
	    job = findJobInQueue(completedJobs, id);
	    if (job != null) {
		history = job.getJobResults();
	    }
	}

	return history;
    }

    public boolean runNextTask(final String node) {
	Task t = null;
	QueryID id = null;

	synchronized (runningJobs) {
	    for (QueuedJob j : runningJobs) {
		if (j.allSearchTasksComplete()) {
		    t = j.generateJoinTask();
		    id = j.getJobID();
		    if (sendTaskToNode(node, t, id)) {
			j.setTaskRunning(t.getID(), node);
			return true;
		    }
		} else if (j.hasUnassignedTasks()) {
		    t = j.getNextUnassignedTask();
		    id = j.getJobID();
		    if (sendTaskToNode(node, t, id)) {
			j.setTaskRunning(t.getID(), node);
			return true;
		    }
		}
	    }
	}

	if (t == null && waitingJobs.size() > 0) {
	    QueuedJob newRun = null;

	    synchronized (waitingJobs) {
		newRun = waitingJobs.peek();
	    }

	    t = newRun.getNextUnassignedTask();
	    id = newRun.getJobID();

	    if (sendTaskToNode(node, t, id)) {
		log.info("QueuedJob: " + id.toString() + " started.");

		newRun.setTaskRunning(t.getID(), node);

		synchronized (waitingJobs) {
		    waitingJobs.poll();
		}

		synchronized (runningJobs) {
		    runningJobs.add(newRun);
		}

		return true;
	    }
	}

	return false;
    }

    public void addTaskResults(final QueryID id, final TaskResults results) {
	QueuedJob job = findJobInQueue(runningJobs, id);
	job.addTaskResults(results);
    }

    public void addJobResults(final QueryID id, final QueryResults results) {
	QueuedJob job = findJobInQueue(runningJobs, id);

	synchronized (runningJobs) {
	    runningJobs.remove(job);
	}

	job.setJobResults(results);

	synchronized (completedJobs) {
	    completedJobs.add(job);
	}

	log.info("QueuedJob: " + id.toString() + " completed.");
    }

    public void pruneCompletedJobs() {
	long limit = 1000L * 60L * 60L * Long.parseLong(configProps
		.getProperty("KeepResultsPeriod"));
	long currentTime = new Date().getTime();

	synchronized (completedJobs) {
	    for (QueuedJob j : completedJobs) {
		if (j.getEndTime().getTime() + limit < currentTime) {
		    log.info(j.getJobID().toString() + " pruned.");
		    completedJobs.remove(j);
		}
	    }
	}
    }

    public void rescheduleTasksOnNode(final String node) {
	synchronized (runningJobs) {
	    for (QueuedJob j : runningJobs) {
		j.checkAndRescheduleTask(node);
	    }
	}
    }

    private QueuedJob findJobInQueue(final Queue<QueuedJob> queue, final QueryID id) {
	synchronized (queue) {
	    for (QueuedJob j : queue) {
		if (j.getJobID().toString().equalsIgnoreCase(id.toString())) {
		    return j;
		}
	    }
	}
	return null;
    }

    @SuppressWarnings("resource")
    private boolean sendTaskToNode(final String node, final Task t, final QueryID id) {
	log.info("Attempting to dispatch " + t.getID() + " to " + node);

	int port = Integer.parseInt(configProps.getProperty("WorkerPort"));

	ObjectOutputStream oos = null;
	PrintWriter out = null;
	BufferedReader in = null;

	Socket sock = null;
	try {
	    sock = new Socket(node, port);

	    out = new PrintWriter(sock.getOutputStream(), true);
	    out.println("Request: application/task");

	    in = new BufferedReader(new InputStreamReader(sock.getInputStream()));
	    String response = in.readLine();

	    if (verifyResponse(response)) {
		oos = new ObjectOutputStream(sock.getOutputStream());
		AbstractMap.SimpleEntry<QueryID, Task> pair = new SimpleEntry<QueryID, Task>(id, t);
		oos.writeObject(pair);

		return true;
	    } else {
		if (sock != null) {
		    sock.close();
		}
		return false;
	    }
	} catch (UnknownHostException e) {
	    log.error(e.getMessage(), e);
	    return false;
	} catch (IOException e) {
	    log.error(e.getMessage(), e);
	    return false;
	}
    }

    private boolean verifyResponse(final String response) {
	if (response != null && response.contains("OK")) {
	    return true;
	}
	return false;
    }
}
